package com.atguigu.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // ExecutionGraph
        // +--------------------+        +----------------------+
        // | source             |  ----> | reduce => print[1]   |
        // +--------------------+    |   +----------------------+
        //                           |-> +----------------------+
        //                               | reduce => print[2]   +
        //                               +----------------------+
        //                           |-> +----------------------+
        //                               | reduce => print[3]   +
        //                               +----------------------+
        //                           |-> +----------------------+
        //                               | reduce => print[4]   +
        //                               +----------------------+
        //
        env
                .fromElements(1,2,3,4,5,6,7,8)
                .setParallelism(1)
                .keyBy(r -> r % 3)
                .reduce((r1, r2) -> r1 + r2)
                .setParallelism(4)
                .print()
                .setParallelism(4);

        env.execute();
    }
}
